package net.bwie.realtime.dws.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.dws.log.bean.MonitorInfo;
import net.bwie.realtime.dws.log.bean.Violation;
import net.bwie.realtime.dws.log.function.MonitorLimitSpeedCache;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import utils.KafkaUtil;

import java.time.Duration;
import java.util.List;
import java.util.Map;


public class DangerousDrivingCEP {


    public static void main(String[] args) throws Exception{

        //1. env-准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        // 增加网络和检查点配置
        env.setParallelism(1);
        env.enableCheckpointing(5000L); // 增加检查点间隔
        env.getCheckpointConfig().setCheckpointTimeout(60000); // 设置检查点超时时间

        //2. source-加载数据  读取kafka中的消息
        DataStream<String> vehicleKafkaStream = KafkaUtil.consumerKafka(env, "dwd-traffic-monitor-log");
//        vehicleKafkaStream.print("kafka");


        //3. transformation-数据处理转换
        DataStream<String> resultStream = mapWindowData(vehicleKafkaStream);
        resultStream.print();

        //5. sink-输出数据


        //4. execute-执行
        env.execute();
    }

    private static DataStream<String> mapWindowData(DataStream<String> vehicleKafkaStream) {

        //转为实体类，并对时间戳进行毫秒级的转化
        SingleOutputStreamOperator<MonitorInfo> monitorinfo = vehicleKafkaStream.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                jsonObject.put("actionTime", jsonObject.getLong("actionTime") * 1000);
                return JSON.parseObject(jsonObject.toJSONString(), MonitorInfo.class);
            }
        });

//        monitorinfo.print();

        // 2. 解析JSON后，提取事件时间并生成水位线
        SingleOutputStreamOperator<MonitorInfo> withWatermarkDs = monitorinfo
                // 提取事件时间（从MonitorInfo的actionTime字段获取，单位：毫秒）
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<MonitorInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner((event, timestamp) -> event.getActionTime())
                );

//        withWatermarkDs.print();

        // 定义Pattern
        Pattern<MonitorInfo, MonitorInfo> pattern = Pattern.<MonitorInfo>begin("first", AfterMatchSkipStrategy.skipPastLastEvent())
                .where(new SimpleCondition<MonitorInfo>() {
                    @Override
                    public boolean filter(MonitorInfo value) throws Exception {
                        Integer limitSpeed = MonitorLimitSpeedCache.loadFromDB(value.getMonitorId());
                        return value.getSpeed() >= limitSpeed * 1.1;
                    }
                }).times(3)
                // 此处的10分钟，一定数据要触发该10分钟的时候才会有结果
                .within(Time.minutes(2));


        // 使用cep
        //在数据流上用模式匹配
        PatternStream<MonitorInfo> patternStream = CEP.pattern(withWatermarkDs.keyBy(v -> v.getCar()), pattern);

        // 输出数据
        //选择输出数据
        SingleOutputStreamOperator<String> ds3 = patternStream.select(new PatternSelectFunction<MonitorInfo, String>() {
            @Override
            public String select(Map<String, List<MonitorInfo>> map) throws Exception {
                // String userId = map.get("first").get(0).getUserId();
                // return userId;
                System.out.println("select.......");
                System.out.println(map);
                return map.get("first").get(0).getCar();
            }
        });
        ds3.print("超速的车牌号是：");

        SingleOutputStreamOperator<Violation> resultStream = ds3.map(new MapFunction<String, Violation>() {
            @Override
            public Violation map(String s) throws Exception {
                return new Violation(0, s, "超速", System.currentTimeMillis());
            }
        });
//        map.print();

        SingleOutputStreamOperator<String> result = resultStream.map(villation -> {
            return String.format(
                    "{\"car\":\"%s\",\"create_time\":\"%s\",\"violation\":\"%s\"}",
                    villation.getCar() != null ? villation.getCar() : "",
                    villation.getCreateTime() != null ? villation.getCreateTime() : "",
                    villation.getViolation() != null ? villation.getViolation() : ""
            );
        });

        return result;
    }
}
